package org.xq.softcup.mq.client.consumer;

import lombok.extern.slf4j.Slf4j;
import org.xq.softcup.mq.client.broker.IMqBroker;
import org.xq.softcup.mq.client.util.MqResult;
import org.xq.softcup.mq.client.consumer.annotation.MqConsumer;
import org.xq.softcup.mq.client.factory.MqClientFactory;
import org.xq.softcup.mq.client.message.MqMessage;
import org.xq.softcup.mq.client.message.MqMessageStatus;
import org.xq.softcup.mq.client.util.LogUtil;
import org.xq.softcup.mq.client.util.ThrowableUtil;
import org.xq.softcup.mq.util.IpUtil;

import java.util.List;
import java.util.UUID;
import java.util.concurrent.Callable;
import java.util.concurrent.FutureTask;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.TimeoutException;
import java.util.concurrent.atomic.AtomicLong;

/**
 * 消费者线程
 * Created by Admin.
 */
@Slf4j
public class ConsumerThread extends Thread {

    private IMqConsumer consumerHandler;
    private MqConsumer mqConsumer;

    private String uuid;

    public ConsumerThread(IMqConsumer consumerHandler) {
        this.consumerHandler = consumerHandler;
        this.mqConsumer = consumerHandler.getClass().getAnnotation(MqConsumer.class);

        this.uuid = UUID.randomUUID().toString().replaceAll("-", "");
    }

    public MqConsumer getMqConsumer() {
        return mqConsumer;
    }

    public String getUuid() {
        return uuid;
    }

    @Override
    public void run() {
        int sleep = 0;
        while (!MqClientFactory.clientFactoryPoolStoped) {
            try {
                // 检查在线的消费者
                ConsumerRegistrySupport.ActiveInfo activeInfo = MqClientFactory.getConsumerRegistrySupport().isActice(this);
                log.debug("检查消费者活动信息, topic:{}, group:{}, ActiveInfo={}", mqConsumer.topic(), mqConsumer.group(), activeInfo);
                if (activeInfo != null) {
                    IMqBroker mqBroker = MqClientFactory.getMqBroker();
                    // 取100条新消息
                    List<MqMessage> messageList = mqBroker.pullNewMessage(mqConsumer.topic(), mqConsumer.group(), activeInfo.rank, activeInfo.total, 50);
                    if (messageList != null && messageList.size() > 0) {
                        // 重置等待时间
                        if (mqConsumer.transaction()) {
                            sleep = 0;    // 事务消息状态被锁定及时更新，不会重试
                        } else {
                            sleep = 1;    // 非事务消息状态被回调延迟更新，会重试，需要等待回调更新状态
                        }
                        //消費消息
                        for (final MqMessage msg : messageList) {
                            // 检查2次在线的消费者
                            ConsumerRegistrySupport.ActiveInfo newActiveInfo = MqClientFactory.getConsumerRegistrySupport().isActice(this);
                            if (!(newActiveInfo != null && newActiveInfo.rank == activeInfo.rank && newActiveInfo.total == activeInfo.total)) {
                                break;
                            }
                            // 锁定消息
                            /*if (mqConsumer.transaction()) {
                                String appendLog_lock = LogUtil.makeLog( "锁定消息", ("消费者信息=" + newActiveInfo.toString() + "；<br>消费者IP=" + IpUtil.getIp()));
                                //锁定消费消息
                                int lockRet = MqClientFactory.getMqBroker().lockMessage(msg.getId(), appendLog_lock);
                                if (lockRet < 1) {
                                    continue;
                                }
                            }*/

                            // 消费消息
                            MqResult mqResult = null;
                            try {
                                if (msg.getTimeout() > 0) {
                                    // 限制超时
                                    Thread futureThread = null;
                                    try {
                                        FutureTask<MqResult> futureTask = new FutureTask<MqResult>(new Callable<MqResult>() {
                                            @Override
                                            public MqResult call() throws Exception {
                                                log.info("Doing consume() >> "+ msg.getData());
                                                //消费者消费消息
                                                return consumerHandler.consume(msg.getData());
                                            }
                                        });
                                        futureThread = new Thread(futureTask);
                                        futureThread.start();
                                        mqResult = futureTask.get(msg.getTimeout(), TimeUnit.SECONDS);
                                    } catch (TimeoutException e) {
                                        log.error(e.getMessage(), e);
                                        mqResult = new MqResult(MqResult.FAIL_CODE, "Timeout:" + e.getMessage());
                                    } finally {
                                        futureThread.interrupt();
                                    }
                                } else {
                                    log.debug("直接消费 Consumer:{} ID:{} ", consumerHandler.getClass().getSimpleName() ,msg.getId());
                                    // 直接执行
                                    mqResult = consumerHandler.consume(msg.getData());
                                }

                                if (mqResult == null) {
                                    mqResult = MqResult.FAIL;
                                }
                            } catch (Exception e) {
                                log.error(e.getMessage(), e);
                                String errorMsg = ThrowableUtil.toString(e);
                                mqResult = new MqResult(MqResult.FAIL_CODE, errorMsg);
                            }

                            // 回调通知成功与否
                            msg.setStatus(mqResult.isSuccess() ? MqMessageStatus.SUCCESS.name() : MqMessageStatus.FAIL.name());
                            //发送回调消息
                            MqClientFactory.callbackMessage(msg);
                            log.debug("消息已完成,  消息日志: ActiveInfo={}", activeInfo.toString());
                        }
                    } else {
                        sleep = 2;//没有消息，睡眠2秒 ;
                    }
                } else {
                    sleep = 5; //没有在线的消费者，休息5秒
                }
            } catch (Exception e) {
                if (!MqClientFactory.clientFactoryPoolStoped) {
                    log.error(e.getMessage(), e);
                }
            }
            // 睡眠
            try {
                TimeUnit.SECONDS.sleep(sleep);
            } catch (Exception e) {
                if (!MqClientFactory.clientFactoryPoolStoped) {
                    log.error(e.getMessage(), e);
                }
            }

        }
    }
}
